
package com.bangcommunity.bbframe.client.zookeeper;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.bangcommunity.bbframe.common.AbstractLifecycle;
import com.bangcommunity.bbframe.common.base.MainConfigLoader;
import com.bangcommunity.bbframe.common.utils.lang.StringUtils;

/**
 * zk客户端
 * <p/>
 * 创建时间: 14-8-6 下午1:52<br/>
 *
 * @author tanghc
 * @since v0.0.1
 */
public class ZKClient extends AbstractLifecycle {

    public static final Logger logger = LoggerFactory.getLogger(ZKClient.class);

    private CuratorFramework zkClient = null;
    private static ConcurrentMap<String, CuratorFramework> zkClientMap = new ConcurrentHashMap<String, CuratorFramework>();

    private ZKClient() {
        this.start();
    };

    @Override
    protected void doStart() {

        synchronized (ZKClientHolder.class) {
            if (!isStart) {
                isStart = true;
                String url = buildUrl();
                logger.debug("use zk url {} to create client", url);
                zkClient = createAndStartByUrl(url);
                logger.warn("ZKClient start success! {} ", url);
            }
        }
    }

    private String buildUrl() {

        String zkUrl = MainConfigLoader.getInstance().getZkUrl();
        if (StringUtils.isBlank(zkUrl)) {
            throw new RuntimeException("no zk url");
        }
        return zkUrl;
    }

    /**
     * 根据ip获取zk client, 可以直接调用该方法，但不建议 请使用 ZKClientManager 调用
     * 
     * @param ip
     * @return
     */
    public static CuratorFramework create(String url) {

        logger.warn(" start conn zk server {} ", url);

        CuratorFramework newClient = zkClientMap.get(url);
        if (null == newClient) {
            newClient = CuratorFrameworkFactory.newClient(url, new ExponentialBackoffRetry(1000, 3));

            zkClientMap.putIfAbsent(url, newClient);
            newClient = zkClientMap.get(url);
        }
        logger.warn("  conn zk server {} success!", url);
        return newClient;
    }

    /**
     * 根据url获取zk client, 可以直接调用该方法，但不建议 请使用 ZKClientManager 调用
     */
    public static CuratorFramework createAndStartByUrl(String url) {

        CuratorFramework curatorFramework = create(url);
        if (null == curatorFramework) {
            throw new RuntimeException("not create zk client: " + url);
        }
        curatorFramework.start();
        return curatorFramework;
    }

    @Override
    public void stop() {

        if (zkClient != null) {
            zkClient.close();
        }
        // throw new RuntimeException("un implemented");
    }

    private static class ZKClientHolder {

        private static final ZKClient instance = new ZKClient();
    }

    /**
     * 获取zk客户端实例（单例）
     * 
     * @return
     */
    public static CuratorFramework getClient() {

        // 初始化client
        // ZKClientHolder.instance.start();
        ZKClient instance = ZKClientHolder.instance;
        return instance.zkClient;
    }

    public static void main(String[] args) {

        getClient();
    }

}
